package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableConfig, TableEnvironment}

object Demo4WordCount {
  def main(args: Array[String]): Unit = {
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      //指定处理模式
      .inStreamingMode() //流处理模式
      .build()

    //flink sql的执行环境
    val tEnv: TableEnvironment = TableEnvironment.create(settings)

    //增加优化参数
    val configuration: TableConfig = tEnv.getConfig

    configuration.set("table.exec.mini-batch.enabled", "true")
    configuration.set("table.exec.mini-batch.allow-latency", "5 s")
    configuration.set("table.exec.mini-batch.size", "5000")
    configuration.set("table.optimizer.agg-phase-strategy", "TWO_PHASE")


    tEnv.executeSql(
      """
        |CREATE TABLE word (
        |    word STRING
        |) WITH (
        |    'connector' = 'datagen',
        |    'rows-per-second'='10000', -- 每秒生成的数据量
        |    'fields.word.length'='1'
        |)
        |""".stripMargin)

    tEnv.executeSql(
      """
        |CREATE TABLE word_count (
        |    word STRING,-- 字段名和字段类型需要和mysql中表的字段名和类型保持一致
        |    num BIGINT,
        |    PRIMARY KEY (word) NOT ENFORCED
        |) WITH (
        |    'connector' = 'jdbc', -- 连接器类型
        |    'url' = 'jdbc:mysql://master:3306/student',-- jdbc地址
        |    'table-name' = 'word_count', -- 需要手动到mysql中创建表
        |    'username' = 'root',
        |    'password' = '123456'
        |)
        |""".stripMargin)

    tEnv.executeSql(
      """
        |insert into word_count
        |select word,count(1) as num from word
        |group by word
        |""".stripMargin)

  }

}
